package com.dz.state.memstate;

import com.dz.source.CheckpointedSourceExample;
import com.dz.timer.MySourceTuple2;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @ClassName MemStateTest
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/15 17:17
 * @Version 1.0
 */
public class MemSteDemo
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT,8089);
        conf.setInteger("heartbeat.timeout",300000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1,conf);
       //开启checkpoint 每5秒钟进行heckpoint;
        env.getCheckpointConfig().setCheckpointInterval(5000);
        //111.229.240.182
        env.setStateBackend(new FsStateBackend("hdfs://localhost:9000/checkpoint/flink/"));
       DataStreamSource streamSource = env.addSource(new MySourceTuple21());
        streamSource.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).print();

         //streamSource.map(new CheckpointedSourceExample.FailingMapper(20));

        //env.getCheckpointConfig().
        env.execute("mem test");
    }
}
